package com.algo.sns;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.HashSet;
import java.net.URLDecoder;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.algo.utils.InputFileList;
import com.miaozhen.utils.MzSequenceFileInputFormat;

public class GetDa extends Configured implements Tool{
    public static class EtlLogMapper extends Mapper<LongWritable, Text, Text, Text>{
	    String renren = "^http://(www\\.)?renren\\.com/([0-9]+)$";
	    String weibo = "^http://(www\\.)?weibo\\.com/([^/]+)/home";
	    String qq = "^http://user\\.qzone\\.qq\\.com/([0-9]+)/infocenter";
            public void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException{
                    String line = value.toString();
                    String[] items = line.split("[=\\^]");
                    Map<String, String> map = new HashMap<String, String>();
                    for(int i = 1; i < items.length; i = i + 2)
                            map.put(items[i-1], items[i]);

		    StringBuilder sb = new StringBuilder();
                    String da = map.get("da");
                    if(da != null)
			    sb.append(DaDecoder.decode(da, true));
                    
		    String url = map.get("c");
		    if(url == null)	url = map.get("pr");
		    if(url!= null){
			url = URLDecoder.decode(url.replaceAll("%(?![0-9a-fA-F]{2})", "%25"), "UTF-8");
			if(url.matches(renren))	sb.append("^R").append(url.replaceAll(renren, "$2"));
			else if(url.matches(weibo))	sb.append("^W").append(url.replaceAll(weibo, "$2"));
			else if(url.matches(qq))	sb.append("^Q").append(url.replaceAll(qq, "$1"));
		    }

		    if(sb.length() > 0){
			String uuid = map.get("uuid");
			context.write(new Text(uuid), new Text(sb.toString()));
		    }
            }
    }
    public static class EtlLogDaMapper extends Mapper<LongWritable, Text, Text, NullWritable>{
            String renren = "^http://(www\\.)?renren\\.com/([0-9]+)$";
            String weibo = "^http://(www\\.)?weibo\\.com/([^/]+)/home";
            String qq = "^http://user\\.qzone\\.qq\\.com/([0-9]+)/infocenter";
            public void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException{
                    String line = value.toString();
                    String[] items = line.split("[=\\^]");
                    Map<String, String> map = new HashMap<String, String>();
                    for(int i = 1; i < items.length; i = i + 2)
                            map.put(items[i-1], items[i]);

                    StringBuilder sb = new StringBuilder();
                    String da = map.get("da");
                    if(da != null)
                            sb.append(DaDecoder.decode(da, true));

                    String url = map.get("c");
                    if(url == null)     url = map.get("pr");
                    if(url!= null){
                        url = URLDecoder.decode(url.replaceAll("%(?![0-9a-fA-F]{2})", "%25"), "UTF-8");
                        if(url.matches(renren)) sb.append("^R").append(url.replaceAll(renren, "$2"));
                        else if(url.matches(weibo))     sb.append("^W").append(url.replaceAll(weibo, "$2"));
                        else if(url.matches(qq))        sb.append("^Q").append(url.replaceAll(qq, "$1"));
			else sb.append("^");
                    }else	sb.append("^");

                    if(sb.length() > 1){
                        String uuid = map.get("uuid");
			String time = map.get("ti");
			String caid = map.get("p");
                        context.write(new Text(uuid+"^"+time+"^"+sb.toString()+"^"+caid), NullWritable.get());
                    }
            }
    }

    public static class DaReducer extends Reducer<Text, Text, Text, NullWritable>{
            public void reduce(Text key, Iterable<Text> values, Context context){
		    Set<String> daSns = new HashSet<String>();
		    Set<String> urlSns = new HashSet<String>();
                    for(Text value : values){
                            String[] daUrl = value.toString().split("\\^");
                            for(String sns : daUrl[0].split(","))
				if(sns.length() > 0)	daSns.add(sns);
			    if(daUrl.length > 1)
				for(String sns : daUrl[1].split(","))
					if(sns.length() > 0)      urlSns.add(sns);
                    }
		    StringBuilder sb = new StringBuilder();
		    sb.append(key.toString()).append("^");
		    for(String sns : daSns)
			sb.append(sns).append(",");
		    sb.setCharAt(sb.length()-1, '^');
		    for(String sns : urlSns)
			sb.append(sns).append(",");
		    sb.deleteCharAt(sb.length()-1);
                    try {
			if(sb.length() > 0)		context.write(new Text(sb.toString()), NullWritable.get());
				} catch (Exception e) {
					throw new RuntimeException(e);
				}
            }
    }
    public int run(String[] args) throws IOException, ClassNotFoundException, InterruptedException{
	    Configuration conf = getConf();
	    if(args.length >= 4)
	    	conf.set("miaozhen.seq.meta", args[3]);
            Job job = Job.getInstance(conf, "getda");
	    job.setJarByClass(GetDa.class);
            List<String> files =  new InputFileList(args[0], args[1]).getEtlLogInputFileList();
            for(String file : files)
                    MultipleInputs.addInputPath(job, new Path(file), MzSequenceFileInputFormat.class, EtlLogDaMapper.class);
            FileOutputFormat.setOutputPath(job, new Path(args[2]));

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(NullWritable.class);

            job.setNumReduceTasks(0);
            return job.waitForCompletion(true) ? 1 : 0;
    }
    public static void main(String[] args){
            try {
				ToolRunner.run(new Configuration(), new GetDa(), args);
			} catch (Exception e) {
				throw new RuntimeException(e);
			}
    }
}

